package com.alibaba.citrus.cr.mn.order.center.autofulfillmentorder.service.impl;

import com.alibaba.citrus.cr.common.annotation.FacadeInvoker;
import com.alibaba.citrus.cr.common.enums.ordercenter.AsyncMessageEventEnum;
import com.alibaba.citrus.cr.common.enums.ordercenter.AutoFulfillmentOrderStatusCirculationEnum;
import com.alibaba.citrus.cr.common.enums.ordercenter.AutoFulfillmentOrderStatusEnum;
import com.alibaba.citrus.cr.common.enums.ordercenter.InnerMessageEventEnum;
import com.alibaba.citrus.cr.common.exception.FunctionException;
import com.alibaba.citrus.cr.common.lock.SaleRedisLock;
import com.alibaba.citrus.cr.mn.order.center.autofulfillmentorder.ability.AutoFulfillmentOrderAbility;
import com.alibaba.citrus.cr.mn.order.center.autofulfillmentorder.service.AutoFulfillmentOrderBizQueryService;
import com.alibaba.citrus.cr.mn.order.center.autofulfillmentorder.service.AutoFulfillmentOrderBizWriteService;
import com.alibaba.citrus.cr.mn.order.center.autofulfillmentorder.service.AutoFulfillmentOrderService;
import com.alibaba.citrus.cr.mn.order.center.common.constants.RedisLockConstants;
import com.alibaba.citrus.cr.mn.order.center.common.util.AssertUtils;
import com.alibaba.citrus.cr.mn.order.center.facade.function.common.SendMsgService;
import com.alibaba.citrus.cr.mn.order.center.facade.kafka.KafkaUtils;
import com.alibaba.citrus.cr.mn.order.center.forecastorder.ForecastOrderBizDomainService;
import com.alibaba.citrus.cr.mn.order.center.forecastorder.ability.ForecastOrderAbility;
import com.alibaba.citrus.cr.mn.order.center.forecastorder.entity.ForecastOrderEntity;
import com.alibaba.citrus.cr.mn.order.center.forecastorder.factory.ForecastOrderFactory;
import com.alibaba.citrus.cr.mn.order.center.forecastorder.service.ForecastOrderBizWriteService;
import com.alibaba.cz.base.tool.log.Log;
import com.alibaba.cz.base.tool.trace.TraceUtil;
import com.alibaba.cz.base.tool.trace.thread.CustomTraceThreadLocal;
import com.alibaba.cz.base.tool.trace.thread.TraceParam;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.lattice2.epoch.db.service.CustomSequenceAccessor;
import com.alibaba.lattice2.epoch.middleware.mq.runtime.EpochMessageService;
import com.alibaba.lattice2.epoch.sdk.model.Result;
import com.epoch.app.bcorder.contants.BcOrderConstants;
import com.epoch.app.crb2btradecenter.messagetask.model.MessageTask;
import com.epoch.app.crb2btradecenter.messagetask.service.MessageTaskService;
import com.epoch.app.crb2btradecenter.model.dto.MessageTaskLoadListRequest;
import com.epoch.app.crb2btradecenter.model.enums.MessageTaskTypeEnum;
import com.epoch.app.crforecastsales.bo.ForecastOrderBO;
import com.epoch.app.crforecastsales.bo.ForecastOrderLineBO;
import com.epoch.app.crforecastsales.domian.cfsbcautofulfillmentorderrecords.dto.CountRequest;
import com.epoch.app.crforecastsales.domian.cfsbcautofulfillmentorderrecords.dto.QueryForListRequest;
import com.epoch.app.crforecastsales.domian.cfsbcautofulfillmentorderrecords.model.CfsBcAutoFulfillmentOrderRecords;
import com.epoch.app.mnordercenter.autofulfillmentorderbiz.dto.RetryAutoFulfillmentOrderRequest;
import com.epoch.app.mnordercenter.contants.MnOrderCenterConstants;
import com.epoch.app.mnordercenter.model.dto.*;
import com.epoch.app.mnordercenter.orderbaserequest.OrderAtomBatchSubmitRequest;
import com.epoch.app.mnordercenter.orderbaserequest.OrderAtomSubmitRequest;
import com.epoch.app.mnordercenter.salesorderatom.service.SalesOrderAtomService;
import com.epoch.app.mnordercenter.service.MnOrderCenterService;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toList;

/**
 * @author baizhuo
 * @version 1.0
 * @date 2023/8/17 20:03
 */
@Service
public class AutoFulfillmentOrderServiceImpl implements AutoFulfillmentOrderService {

    private static final Log log = Log.getLogger(AutoFulfillmentOrderService.class);

    @Resource
    private ForecastOrderBizWriteService forecastOrderBizWriteService;

    @Resource
    private ForecastOrderBizDomainService forecastOrderBizDomainService;

    @Resource
    private AutoFulfillmentOrderBizQueryService autoFulfillmentOrderBizQueryService;

    @Resource
    private AutoFulfillmentOrderBizWriteService autoFulfillmentOrderBizWriteService;

    @Resource
    private CustomSequenceAccessor customSequenceAccessor;

    @Resource
    private MessageTaskService messageTaskService;

    @Resource
    private EpochMessageService messageService;

    @Resource(name = "autoFulfillmentTaskExecutor")
    private ThreadPoolTaskExecutor autoFulfillmentTaskExecutor;

    @Resource
    private SalesOrderAtomService salesOrderAtomService;

    @Resource
    private MnOrderCenterService mnOrderCenterService;

    @Resource
    private AutoFulfillmentOrderAbility autoFulfillmentOrderAbility;

    @Resource
    private SendMsgService sendMsgService;

    private static final List<Integer> processingCirculationStatusEnumList = Arrays.asList(AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_SUCCEED.getCode(), AutoFulfillmentOrderStatusCirculationEnum.CREATE_SALE_ORDER_SUCCEED.getCode());

    @Override
    @FacadeInvoker
    public Result<ForecastBizAsyncResponse> createAutoFulfillmentOrders(ForecastOrderBizBatchCreateRequest request) {
        log.info("批量创建预报单开始create_auto_fulfillment_orders start,param:{}", JSON.toJSONString(request));
        ForecastBizAsyncResponse createResponse = new ForecastBizAsyncResponse();
        List<ForecastOrderEntity> forecastOrderEntities = ForecastOrderFactory.createDraft(request);
        ForecastOrderAbility.paramsNotBlankCheck(forecastOrderEntities);
        ForecastOrderAbility.paramsLengthCheck(forecastOrderEntities);
        AutoFulfillmentOrderAbility.subOrderNumCheck(forecastOrderEntities);
        autoFulfillmentOrderAbility.paramsEnumDictCheckFromDictionaryService(forecastOrderEntities);
        String sequenceNo = customSequenceAccessor.getNextStringSequence("cr_b2b_trade_center", "batchCreateOrderSequence");
        //请求幂等
        String lockKey = RedisLockConstants.FORECAST_ORDER_BIZ_PREFIX + request.getFromSys() + "#" + request.getReqNo();
        try (SaleRedisLock reqLock = new SaleRedisLock(RedisLockConstants.CACHE_CODE, lockKey, RedisLockConstants.TIMEOUT_MSECS, RedisLockConstants.EXPIRE_MSECS, RedisLockConstants.SLEEP_MILLIS)) {
            if (!reqLock.lock()) {
                log.error("AutoFulfillmentOrderServiceImpl#createAutoFulfillmentOrders 分布式锁获取失败 lockKey={}", lockKey);
                return Result.fail("OC-00-000-01-99-002", "分布式锁获取失败");
            }
            Boolean reqIdempotentProcess = reqIdempotentProcess(request.getReqNo(), request.getFromSys(), sequenceNo, createResponse);
            if (!reqIdempotentProcess) {
                log.info("批量创建预报单，任务幂等查询结果:{}", JSON.toJSONString(createResponse));
                return Result.success(createResponse);
            }
        } catch (Exception e) {
            log.error("批量创建预报单异常", e);
            return Result.fail("OC-00-000-01-99-002", "获取来源单号分布式锁异常");
        }
        List<ForecastOrderBO> createForecastOrderBOS = forecastOrderEntities.stream().map(ForecastOrderEntity::getForecastOrderBO).collect(Collectors.toList());
        String traceId = TraceUtil.getTraceId();
//        String tenantId = TenantUtil.getTenantId();
        //异步执行
        autoFulfillmentTaskExecutor.execute(() -> {
            try {
                TraceParam traceParam = new TraceParam();
                traceParam.setTraceId(traceId);
//                traceParam.setTenantId(tenantId);
                CustomTraceThreadLocal.set(traceParam);
                this.batchCreateAutoFulfillmentOrder(createForecastOrderBOS, sequenceNo, request.getFromSys(),request.getBusinessType(), request.getBusinessUnitCode());
            } catch (Exception e) {
                log.error("async execute create order error", e);
            } finally {
                CustomTraceThreadLocal.clean();
            }
        } );
        createResponse.setBatchTaskId(sequenceNo);
        log.info("批量创建预报单结束create_auto_fulfillment_orders end, result:{}", JSON.toJSONString(createResponse));
        return Result.success(createResponse);
    }

    /**
     * 校验子单数量
     * @param forecastOrderEntities
     */
    private void subOrderNumCheck(List<ForecastOrderEntity> forecastOrderEntities) {
        for (ForecastOrderEntity forecastOrderEntity : forecastOrderEntities) {
            AssertUtils.isTrue(forecastOrderEntity.getForecastOrderBO().getForecastOrderLineBOList().size() > 10, "OC-01-001-01-15-004", "forecastOrderDetailDTOS", forecastOrderEntity.getForecastOrderBO().getForecastOrderLineBOList().size(), "10");
        }
    }

    /**
     * 自动履约-批量创建预报订单
     * @param forecastOrderBOS 预报单bo
     * @param taskId 批次号
     * @param fromSys 来源系统编码
     * @param businessUnitCode 业务单元编码
     */
    private void batchCreateAutoFulfillmentOrder(List<ForecastOrderBO> forecastOrderBOS, String taskId, String fromSys, String businessType, String businessUnitCode) {
        log.info("异步批量创建预报单开始, forecastOrderBOS:{}, taskId:{}, fromSys:{}, businessUnitCode:{}", JSON.toJSONString(forecastOrderBOS), taskId, fromSys, businessUnitCode);
        try {
            for (ForecastOrderBO forecastOrder : forecastOrderBOS) {
                if (null != forecastOrder.getFeatures()) {
                    forecastOrder.getFeatures().put("automaticOrder", "true");
                } else {
                    Map map = new HashMap();
                    map.put("automaticOrder", "true");
                    forecastOrder.setFeatures(map);
                }
            }
            //复用之前的方法，根据CreateForecastResponseDTO状态，从forecastOrderBOS获取预报单号及状态保存到自动履约订单表
            List<CreateForecastResponseDTO> createForecastResponseDTOS = new ArrayList<>();
            boolean createForecastOrderFailed = false;
            try {
                createForecastResponseDTOS = forecastOrderBizWriteService.batchCreateWithReturn(forecastOrderBOS, taskId, fromSys, businessUnitCode);
            } catch (Exception e1) {
                log.error("批量创建带返回值执行失败", e1);
                createForecastOrderFailed = true;
            }
            Map<String, CreateForecastResponseDTO> forecastResponseDTOMap = createForecastResponseDTOS.stream().collect(Collectors.toMap(CreateForecastResponseDTO::getSourceOrderNo, Function.identity(), (k1, k2) -> k2));
            List<CfsBcAutoFulfillmentOrderRecords> createForecastOrderSuccessList = new ArrayList<>();
            for (ForecastOrderBO forecastOrderBO : forecastOrderBOS) {
                for (ForecastOrderLineBO forecastOrderLineBO : forecastOrderBO.getForecastOrderLineBOList()) {
                    //判断自动履约订单表是否已存在该来源单号
                    QueryForListRequest queryForListRequest = new QueryForListRequest();
                    queryForListRequest.setSourceOrderNo(forecastOrderBO.getSourceOrderNo());
                    queryForListRequest.setSourceOrderLineNo(forecastOrderLineBO.getSourceLineNum());
                    queryForListRequest.setFromSys(forecastOrderBO.getFromSys());
                    queryForListRequest.setBusinessUnitCode(forecastOrderBO.getBusinessUnitCode());
                    queryForListRequest.setBusinessType(forecastOrderBO.getBusinessType());
                    log.info("查询自动履约订单开始 param:{}", JSON.toJSONString(queryForListRequest));
                    List<CfsBcAutoFulfillmentOrderRecords> existCfsBcAutoFulfillmentOrderRecordList = autoFulfillmentOrderBizQueryService.queryForList(queryForListRequest).stream().filter(record -> record.getCirculationStatus() > AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getCode()).collect(Collectors.toList());
                    log.info("查询自动履约订单结束并过滤掉创建预报单失败的自动履约订单 result:{}", existCfsBcAutoFulfillmentOrderRecordList);
                    CfsBcAutoFulfillmentOrderRecords newCfsBcAutoFulfillmentOrderRecords = new CfsBcAutoFulfillmentOrderRecords();
                    newCfsBcAutoFulfillmentOrderRecords.setBatchNo(taskId);
                    newCfsBcAutoFulfillmentOrderRecords.setReqNo(forecastOrderBO.getReqNo());
                    newCfsBcAutoFulfillmentOrderRecords.setSourceOrderNo(forecastOrderBO.getSourceOrderNo());
                    newCfsBcAutoFulfillmentOrderRecords.setSourceOrderLineNo(forecastOrderLineBO.getSourceLineNum());
                    newCfsBcAutoFulfillmentOrderRecords.setFromSys(forecastOrderBO.getFromSys());
                    newCfsBcAutoFulfillmentOrderRecords.setBusinessType(forecastOrderBO.getBusinessType());
                    newCfsBcAutoFulfillmentOrderRecords.setBusinessUnitCode(forecastOrderBO.getBusinessUnitCode());
                    newCfsBcAutoFulfillmentOrderRecords.setRetryNum(0);
                    newCfsBcAutoFulfillmentOrderRecords.setVersion(1);
                    newCfsBcAutoFulfillmentOrderRecords.setGmt_create(new Date());
                    newCfsBcAutoFulfillmentOrderRecords.setGmt_modified(new Date());
                    newCfsBcAutoFulfillmentOrderRecords.setCreator(forecastOrderBO.getCreatorCode());
                    newCfsBcAutoFulfillmentOrderRecords.setUpdater(forecastOrderBO.getCreatorCode());
                    if (existCfsBcAutoFulfillmentOrderRecordList.size() >= 1) {
                        newCfsBcAutoFulfillmentOrderRecords.setStatus(AutoFulfillmentOrderStatusEnum.FAIL.getCode());
                        newCfsBcAutoFulfillmentOrderRecords.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getCode());
                        newCfsBcAutoFulfillmentOrderRecords.setFailReason("来源单号及行号重复");
                    } else {
                        if (forecastResponseDTOMap.containsKey(forecastOrderBO.getSourceOrderNo())) {
                            CreateForecastResponseDTO createForecastResponseDTO = forecastResponseDTOMap.get(forecastOrderBO.getSourceOrderNo());
                            if (!createForecastResponseDTO.getSuccess() || createForecastOrderFailed || StringUtils.isEmpty(createForecastResponseDTO.getSalesOrderNo())) {
                                newCfsBcAutoFulfillmentOrderRecords.setStatus(AutoFulfillmentOrderStatusEnum.FAIL.getCode());
                                newCfsBcAutoFulfillmentOrderRecords.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getCode());
                                newCfsBcAutoFulfillmentOrderRecords.setFailReason(StringUtils.isEmpty(createForecastResponseDTO.getMessage()) ? AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getDesc() : createForecastResponseDTO.getMessage());
                            } else {
                                newCfsBcAutoFulfillmentOrderRecords.setStatus(AutoFulfillmentOrderStatusEnum.PROCESSING.getCode());
                                newCfsBcAutoFulfillmentOrderRecords.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_SUCCEED.getCode());
                                newCfsBcAutoFulfillmentOrderRecords.setOrderBizNo(createForecastResponseDTO.getSalesOrderNo());
                                newCfsBcAutoFulfillmentOrderRecords.setOrderLineBizNo(forecastOrderLineBO.getLineNum());
                            }
                        }
                    }
                    int lastInsertId = autoFulfillmentOrderBizWriteService.create(newCfsBcAutoFulfillmentOrderRecords);
                    newCfsBcAutoFulfillmentOrderRecords.setId((long) lastInsertId);
                    if (newCfsBcAutoFulfillmentOrderRecords.getCirculationStatus() == AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_SUCCEED.getCode()) {
                        createForecastOrderSuccessList.add(newCfsBcAutoFulfillmentOrderRecords);
                    }
                }
            }
            if (CollectionUtils.isNotEmpty(createForecastOrderSuccessList)) {
                //创建（预报订单）成功的的订单，进行下一步：预报订单分析
                log.info("异步创建预报单，开始批量分析");
                batchAnalyze(createForecastOrderSuccessList);
                // 预报单创建成功通知应用
                List<String> bizOrderNoList = createForecastResponseDTOS.stream().filter(CreateForecastResponseDTO::getSuccess).map(CreateForecastResponseDTO::getSalesOrderNo).collect(toList());
                sendMsgService.sendForecastOrderChangeMsg(bizOrderNoList, InnerMessageEventEnum.FORECAST_ORDER_CREATED);
            } else {
                //全部失败，直接发送Kafka消息
                log.info("发送Kafka消息, taskId:{}", taskId);
                sendMessage(taskId, fromSys, businessType, businessUnitCode);
            }
        } catch (Exception e) {
            log.error("异步批量创建预报单失败", e);
            throw e;
        }
        log.info("异步批量创建预报订单结束");
    }

    /**
     * 自动履约-批量分析预报订单
     * @param cfsBcAutoFulfillmentOrderRecordsList 自动履约订单
     */
    private void batchAnalyze(List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsList) {
        log.info("预报单批量分析开始,param:{}", JSON.toJSONString(cfsBcAutoFulfillmentOrderRecordsList));
        //尽量复用之前代码,调用原来的批量分析预报订单方法
        AnalysisForecastRequest analysisForecastRequest = new AnalysisForecastRequest();
        analysisForecastRequest.setFromSys(cfsBcAutoFulfillmentOrderRecordsList.get(0).getFromSys());
        analysisForecastRequest.setBusinessType(cfsBcAutoFulfillmentOrderRecordsList.get(0).getBusinessType());
        analysisForecastRequest.setBusinessUnitCode(cfsBcAutoFulfillmentOrderRecordsList.get(0).getBusinessUnitCode());
        analysisForecastRequest.setModifierCode(cfsBcAutoFulfillmentOrderRecordsList.get(0).getUpdater());
        List<AnalysisForecastBizDTO> analysisForecastBizDTOList = new ArrayList<>();
        for (CfsBcAutoFulfillmentOrderRecords cfsBcAutoFulfillmentOrderRecords : cfsBcAutoFulfillmentOrderRecordsList) {
            AnalysisForecastBizDTO analysisForecastBizDTO = new AnalysisForecastBizDTO();
            analysisForecastBizDTO.setSalesOrderNo(cfsBcAutoFulfillmentOrderRecords.getOrderBizNo());
            analysisForecastBizDTO.setLineNum(cfsBcAutoFulfillmentOrderRecords.getOrderLineBizNo());
            analysisForecastBizDTOList.add(analysisForecastBizDTO);
        }
        analysisForecastRequest.setAnalysisForecastNoDTOList(analysisForecastBizDTOList);
        log.info("批量分析预报单开始 batchAnalysis :{}", JSON.toJSONString(analysisForecastRequest));
        Result<AnalysisForecastResponseDTO> analyzeResult = forecastOrderBizWriteService.batchAnalysis(analysisForecastRequest);
        log.info("批量分析预报单结束 batchAnalysis result:{}", JSON.toJSONString(analyzeResult));
        boolean analyzeFailed = !analyzeResult.isSuccess() && "OC-04-003-01-16-009".equals(analyzeResult.getErrorCode());
        boolean needSend = false;
        for (CfsBcAutoFulfillmentOrderRecords autoFulfillmentOrderRecord : cfsBcAutoFulfillmentOrderRecordsList) {
            if (analyzeFailed) {
                autoFulfillmentOrderRecord.setStatus(AutoFulfillmentOrderStatusEnum.FAIL.getCode());
                autoFulfillmentOrderRecord.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.ANALYZE_FAILED.getCode());
                autoFulfillmentOrderRecord.setFailReason(analyzeResult.getMessage());
            } else if ((!analyzeResult.isSuccess() && "OC-04-003-01-16-008".equals(analyzeResult.getErrorCode()))) {
                autoFulfillmentOrderRecord.setStatus(AutoFulfillmentOrderStatusEnum.PROCESSING.getCode());
                needSend = true;
            } else if (analyzeResult.isSuccess()) {
                autoFulfillmentOrderRecord.setStatus(AutoFulfillmentOrderStatusEnum.PROCESSING.getCode());
                autoFulfillmentOrderRecord.setAnalyzeBatchNo(analyzeResult.getResult().getBatchNo());
                needSend = true;
            }
            autoFulfillmentOrderRecord.setGmt_modified(new Date());
            autoFulfillmentOrderBizWriteService.update(autoFulfillmentOrderRecord);
        }
        //因为是等待该批次号所有可分析订单分析完成再进行下一步，故一个批次号只发一条消息即可
        if (needSend) {
            //发送rocketMQ消息，自发自收，目的是等所有预报订单分析完毕
            String message = JSON.toJSONString(cfsBcAutoFulfillmentOrderRecordsList.get(0));
            Map<String, String> map = new HashMap<>(1);
            map.put("_CZ_KEY", BcOrderConstants.TENANT_ID());
            Boolean sendResult = messageService.sendMessage("auto_fulfillment_message", "CR_AUTO_FULFILLMENT_ORDER", "AUTO_FULFILLMENT_ORDER_ANALYZE", cfsBcAutoFulfillmentOrderRecordsList.get(0).getBatchNo(), message, map);
            log.info("AutoFulfillmentOrderServiceImpl#batchAnalyze AUTO_FULFILLMENT_ORDER_ANALYZE message:{}, sendResult:{}", message, sendResult);
        } else if (analyzeFailed) {
            // 分析失败，发送Kafka消息,需要将创建预报订单这一步的错误信息一并发送
            CfsBcAutoFulfillmentOrderRecords cfsBcAutoFulfillmentOrderRecords = cfsBcAutoFulfillmentOrderRecordsList.get(0);
            log.info("batchAnalyze analyzeFailed, send kafka message. ");
            sendMessage(cfsBcAutoFulfillmentOrderRecords.getBatchNo(), cfsBcAutoFulfillmentOrderRecords.getFromSys(), cfsBcAutoFulfillmentOrderRecords.getBusinessType(), cfsBcAutoFulfillmentOrderRecords.getBusinessUnitCode());
        }
        log.info("预报单批量分析结束");
    }

    /**
     * 校验修改人
     * @param modifierCode
     * @return
     */
    private Modifiers invalidModifier(String modifierCode) {
        Map<String, Modifiers> modifierMap = new HashMap<>();
        //修改人查询
        BatchQueryEmployeeByConditionRequest modifierRequest = new BatchQueryEmployeeByConditionRequest();
        modifierRequest.setWorkIdList(Collections.singletonList(modifierCode));
        Result<List<Modifiers>> modifierResult = mnOrderCenterService.batchQueryEmployeeByCondition(modifierRequest);
        if (Objects.nonNull(modifierResult) && CollectionUtils.isNotEmpty(modifierResult.getResult())) {
            modifierMap = modifierResult.getResult().stream().collect(Collectors.toMap(Modifiers::getWorkId, Function.identity(), (v1, v2) -> v1));
        }
        //修改人校验
        if (!modifierMap.containsKey(modifierCode)) {
            throw new FunctionException("OC-04-004-01-16-005", modifierCode);
        }
        Modifiers modifier = modifierMap.get(modifierCode);
        if (Objects.nonNull(modifier.getStatus()) && modifier.getStatus() != 1) {
            throw new FunctionException("OC-04-004-01-16-007", modifierCode);
        }
        return modifier;
    }

    /**
     * 请求幂等判断
     * @param reqNo 请求编号
     * @param taskId 批次号
     * @param forecastBizAsyncResponse 响应
     * @return 是否幂等
     */
    private Boolean reqIdempotentProcess(String reqNo, String fromSys, String taskId, ForecastBizAsyncResponse forecastBizAsyncResponse) {
        try {
            //请求幂等
            MessageTaskLoadListRequest loadRequest = new MessageTaskLoadListRequest();
            loadRequest.setReqNo(reqNo);
            loadRequest.setFromSys(fromSys);
            Result<List<MessageTask>> result = messageTaskService.loadList(loadRequest);
            if(Objects.isNull(result) || CollectionUtils.isEmpty(result.getResult())) {
                createMessageTask(reqNo, fromSys, taskId);
                return true;
            }
            forecastBizAsyncResponse.setBatchTaskId(result.getResult().get(0).getBatchNo());
        } catch (Exception e) {
            log.error("AutoFulfillmentOrderServiceImpl#reqIdempotentProcess load error", e);
        }
        return false;
    }

    /**
     * 创建messageTask
     * @param reqNo 请求编号
     * @param innerSequenceNo 序列号
     */
    private void createMessageTask(String reqNo, String fromSys, String innerSequenceNo) {
        MessageTask messageTask = new MessageTask();
        messageTask.setReqNo(reqNo);
        messageTask.setFromSys(fromSys);
        messageTask.setBatchNo(innerSequenceNo);
        messageTask.setType(MessageTaskTypeEnum.OC_FORECAST_ORDER_BATCH_CREATE.getValue());
        messageTask.setGmt_create(new Date());
        messageTask.setGmt_modified(new Date());
        messageTask.setStatus(1);
        messageTaskService.create(messageTask);
    }

    @Override
    @FacadeInvoker
    public Result<RetryAutoFulfillmentOrderResponse> retryAutoFulfillmentOrder(RetryAutoFulfillmentOrderRequest orderBatchRetryRequest) {
        log.info("批量重试创建预报单开始, param:{}", JSON.toJSONString(orderBatchRetryRequest));
        ForecastOrderAbility.baseParamsCheck(orderBatchRetryRequest.getFromSys(), orderBatchRetryRequest.getBusinessType(), orderBatchRetryRequest.getBusinessUnitCode());
        autoFulfillmentOrderAbility.checkOrderBatchRetryRequestParam(orderBatchRetryRequest);
        CountRequest countRequest = new CountRequest();
        countRequest.setFromSys(orderBatchRetryRequest.getFromSys());
        countRequest.setBusinessType(orderBatchRetryRequest.getBusinessType());
        countRequest.setBusinessUnitCode(orderBatchRetryRequest.getBusinessUnitCode());
        countRequest.setOrderBizNos(orderBatchRetryRequest.getOrderBizNos());
        int count = autoFulfillmentOrderBizQueryService.count(countRequest);
        if (1000 < count) {
            throw new FunctionException("OC-04-001-01-16-034");
        }
        RetryAutoFulfillmentOrderResponse orderBatchRetryResponse = new RetryAutoFulfillmentOrderResponse();
        String sequenceNo = customSequenceAccessor.getNextStringSequence("cr_b2b_trade_center", "batchCreateOrderSequence");
        orderBatchRetryResponse.setBatchNo(sequenceNo);
        List<CfsBcAutoFulfillmentOrderRecords> needRetryList = new ArrayList<>();
        //请求幂等
        String lockKey = RedisLockConstants.FORECAST_ORDER_BIZ_PREFIX + orderBatchRetryRequest.getFromSys() + "#" + orderBatchRetryRequest.getReqNo();
        try (SaleRedisLock reqLock = new SaleRedisLock(RedisLockConstants.CACHE_CODE, lockKey, RedisLockConstants.TIMEOUT_MSECS, RedisLockConstants.EXPIRE_MSECS, RedisLockConstants.SLEEP_MILLIS)) {
            if (!reqLock.lock()) {
                log.error("AutoFulfillmentOrderServiceImpl#retryAutoFulfillmentOrder 分布式锁获取失败 lockKey={}", lockKey);
                return Result.fail("OC-00-000-01-99-002", "分布式锁获取失败");
            }
            String batchNo = queryBatchNoByReqNo(orderBatchRetryRequest.getReqNo());
            if (StringUtils.isNotEmpty(batchNo)) {
                //返回老的
                orderBatchRetryResponse.setBatchNo(batchNo);
                return Result.success(orderBatchRetryResponse);
            }
            QueryForListRequest queryForListRequest = new QueryForListRequest();
            queryForListRequest.setOrderBizNos(orderBatchRetryRequest.getOrderBizNos());
            queryForListRequest.setFromSys(orderBatchRetryRequest.getFromSys());
            queryForListRequest.setBusinessType(orderBatchRetryRequest.getBusinessType());
            queryForListRequest.setBusinessUnitCode(orderBatchRetryRequest.getBusinessUnitCode());
            List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsList = autoFulfillmentOrderBizQueryService.queryForList(queryForListRequest).stream().filter(record -> record.getCirculationStatus() > AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getCode()).collect(Collectors.toList());
            Map<String, List<CfsBcAutoFulfillmentOrderRecords>> orderBizNoMap = cfsBcAutoFulfillmentOrderRecordsList.stream().collect(Collectors.groupingBy(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo));
            for (Object orderBizNoObj : orderBatchRetryRequest.getOrderBizNos()) {
                String orderBizNo = (String) orderBizNoObj;
                if (!orderBizNoMap.containsKey(orderBizNo)) {
                    throw new FunctionException("OC-04-004-01-16-003", orderBizNo);
                } else  {
                    for (CfsBcAutoFulfillmentOrderRecords cfsBcAutoFulfillmentOrderRecords : orderBizNoMap.get(orderBizNo)) {
                        //一主多子分析后，可能出现部分子单完全满足、部分子单完全满足但活动冲突的情况，此时完全满足的子单status=1、circulation_status=28
                        if (AutoFulfillmentOrderStatusEnum.PROCESSING.getCode() == cfsBcAutoFulfillmentOrderRecords.getStatus()) {
                            throw new FunctionException("OC-04-004-01-16-001", orderBizNo);
                        } else if (AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()) {
                            throw new FunctionException("OC-04-004-01-16-003", orderBizNo);
                        } else if (AutoFulfillmentOrderStatusEnum.SUCCESS.getCode() == cfsBcAutoFulfillmentOrderRecords.getStatus()
                                || AutoFulfillmentOrderStatusCirculationEnum.PUSH_TO_SAP_SUCCEED.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()) {
                            throw new FunctionException("OC-04-004-01-16-002", orderBizNo);
                        } else {
                            //需重置reqNo
                            cfsBcAutoFulfillmentOrderRecords.setReqNo(orderBatchRetryRequest.getReqNo());
                            cfsBcAutoFulfillmentOrderRecords.setUpdater(orderBatchRetryRequest.getModifierCode());
                            cfsBcAutoFulfillmentOrderRecords.setBatchNo(sequenceNo);
                            cfsBcAutoFulfillmentOrderRecords.setStatus(AutoFulfillmentOrderStatusEnum.PROCESSING.getCode());
                            needRetryList.add(cfsBcAutoFulfillmentOrderRecords);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("批量重试创建预报单异常 error", e);
            throw e;
        }
        String traceId = TraceUtil.getTraceId();
//        String tenantId = TenantUtil.getTenantId();
        //异步执行
        autoFulfillmentTaskExecutor.execute(() -> {
            try {
                TraceParam traceParam = new TraceParam();
                traceParam.setTraceId(traceId);
//                traceParam.setTenantId(tenantId);
                CustomTraceThreadLocal.set(traceParam);
                this.asyncRetry(needRetryList);
            } catch (Exception e) {
                log.error("async execute retry error", e);
            } finally {
                CustomTraceThreadLocal.clean();
            }
        });
        log.info("批量重试创建预报单结束, result:{}", JSON.toJSONString(orderBatchRetryResponse));
        return Result.success(orderBatchRetryResponse);
    }

    /**
     * 根据请求编号查询批次号
     * @param reqNo 请求编号
     * @return 批次号
     */
    private String queryBatchNoByReqNo(String reqNo) {
        QueryForListRequest request = new QueryForListRequest();
        request.setReqNo(reqNo);
        List<CfsBcAutoFulfillmentOrderRecords> queryResult = autoFulfillmentOrderBizQueryService.queryForList(request);
        if (CollectionUtils.isNotEmpty(queryResult)) {
            return queryResult.get(0).getBatchNo();
        }
        return null;
    }

    /**
     * 异步重试
     * @param cfsBcAutoFulfillmentOrderRecordsList 自动履约订单
     */
    private void asyncRetry(List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsList) {
        log.info("异步批量重试开始, param:{}", JSON.toJSONString(cfsBcAutoFulfillmentOrderRecordsList));
        try {
            if (CollectionUtils.isEmpty(cfsBcAutoFulfillmentOrderRecordsList)) {
                return;
            }
            List<CfsBcAutoFulfillmentOrderRecords> needSubmitList = new ArrayList<>();
            List<CfsBcAutoFulfillmentOrderRecords> needAnalyzeList = new ArrayList<>();
            List<CfsBcAutoFulfillmentOrderRecords> needConfirmList = new ArrayList<>();
            List<CfsBcAutoFulfillmentOrderRecords> activityConflictList = new ArrayList<>();
            List<CfsBcAutoFulfillmentOrderRecords> creditNotEnoughList = new ArrayList<>();
            List<CfsBcAutoFulfillmentOrderRecords> needUpdateList = new ArrayList<>();
            List<String> stockNotEnoughOrderBizNoList = new ArrayList<>();
            for (CfsBcAutoFulfillmentOrderRecords cfsBcAutoFulfillmentOrderRecords : cfsBcAutoFulfillmentOrderRecordsList) {
                if (AutoFulfillmentOrderStatusCirculationEnum.CREATE_SALE_ORDER_SUCCEED.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()
                        || AutoFulfillmentOrderStatusCirculationEnum.PUSH_TO_SAP_FAILED.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()) {
                    needSubmitList.add(cfsBcAutoFulfillmentOrderRecords);
                } else if (AutoFulfillmentOrderStatusCirculationEnum.ACTIVITY_CONFLICT.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()
                        || AutoFulfillmentOrderStatusCirculationEnum.STOCK_NOT_ENOUGH_AND_ACTIVITY_CONFLICT.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()
                        || AutoFulfillmentOrderStatusCirculationEnum.CREDIT_NOT_ENOUGH_AND_ACTIVITY_CONFLICT.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()
                        || AutoFulfillmentOrderStatusCirculationEnum.OTHER_REASON_AND_ACTIVITY_CONFLICT.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()) {
                    activityConflictList.add(cfsBcAutoFulfillmentOrderRecords);
                } else if (AutoFulfillmentOrderStatusCirculationEnum.CREDIT_NOT_ENOUGH.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()) {
                    creditNotEnoughList.add(cfsBcAutoFulfillmentOrderRecords);
                } else if (AutoFulfillmentOrderStatusCirculationEnum.ALL_PASS.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()
                        || AutoFulfillmentOrderStatusCirculationEnum.CREATE_SALE_ORDER_FAILED.getCode() == cfsBcAutoFulfillmentOrderRecords.getCirculationStatus()) {
                    needConfirmList.add(cfsBcAutoFulfillmentOrderRecords);
                } else {
                    //不是完全满足，需要再次分析
                    needAnalyzeList.add(cfsBcAutoFulfillmentOrderRecords);
                }
            }
            if (CollectionUtils.isNotEmpty(activityConflictList)) {
                Map<String, List<CfsBcAutoFulfillmentOrderRecords>> autoFulfillmentOrderRecordsMap = activityConflictList.stream().collect(Collectors.groupingBy(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo));
                //重新查询预报订单状态
                List<String> orderBizNoList = new ArrayList<>(activityConflictList.stream().map(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo).collect(Collectors.toSet()));
                List<PageQueryForecastDetailResponse> forecastDetailResponseList = this.pageQueryLine(orderBizNoList, activityConflictList.get(0));
                if ( CollectionUtils.isNotEmpty(forecastDetailResponseList)) {
                    for (PageQueryForecastDetailResponse pageQueryForecastDetailResponse : forecastDetailResponseList) {
                        if (autoFulfillmentOrderRecordsMap.containsKey(pageQueryForecastDetailResponse.getSalesOrderNo())) {
                            List<CfsBcAutoFulfillmentOrderRecords> newActivityConflictStatus = autoFulfillmentOrderRecordsMap.get(pageQueryForecastDetailResponse.getSalesOrderNo());
                            for (CfsBcAutoFulfillmentOrderRecords c : newActivityConflictStatus) {
                                if (c.getOrderLineBizNo().equals(pageQueryForecastDetailResponse.getLineNum())) {
                                    if (pageQueryForecastDetailResponse.getActivityConfilctFlag()) {
                                        //仍冲突，重新分析
                                        needAnalyzeList.add(c);
                                    } else {
                                        //不冲突
                                        if (AutoFulfillmentOrderStatusCirculationEnum.STOCK_NOT_ENOUGH_AND_ACTIVITY_CONFLICT.getCode() == c.getCirculationStatus()
                                                || AutoFulfillmentOrderStatusCirculationEnum.OTHER_REASON_AND_ACTIVITY_CONFLICT.getCode() == c.getCirculationStatus() ) {
                                            needAnalyzeList.add(c);
                                        } else if (AutoFulfillmentOrderStatusCirculationEnum.CREDIT_NOT_ENOUGH_AND_ACTIVITY_CONFLICT.getCode() == c.getCirculationStatus()) {
                                            //重置CirculationStatus
                                            c.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.CREDIT_NOT_ENOUGH.getCode());
                                            c.setFailReason(AutoFulfillmentOrderStatusCirculationEnum.CREDIT_NOT_ENOUGH.getDesc());
                                            creditNotEnoughList.add(c);
                                        } else if (AutoFulfillmentOrderStatusCirculationEnum.ACTIVITY_CONFLICT.getCode() == c.getCirculationStatus()) {
                                            //重置CirculationStatus
                                            c.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.ALL_PASS.getCode());
                                            c.setFailReason("");
                                            needConfirmList.add(c);
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
            if (CollectionUtils.isNotEmpty(creditNotEnoughList)) {
                Map<String, List<CfsBcAutoFulfillmentOrderRecords>> autoFulfillmentOrderRecordsMap = creditNotEnoughList.stream().collect(Collectors.groupingBy(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo));
                //重新查询预报订单状态
                List<String> orderBizNoListOfCreditNotEnough = new ArrayList<>(creditNotEnoughList.stream().map(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo).collect(Collectors.toSet()));
                List<PageQueryForecastDetailResponse> forecastDetailResponseList = this.pageQueryLine(orderBizNoListOfCreditNotEnough, creditNotEnoughList.get(0));
                if (CollectionUtils.isNotEmpty(forecastDetailResponseList)) {
                    for (PageQueryForecastDetailResponse pageQueryForecastDetailResponse : forecastDetailResponseList) {
                        if (autoFulfillmentOrderRecordsMap.containsKey(pageQueryForecastDetailResponse.getSalesOrderNo())) {
                            List<CfsBcAutoFulfillmentOrderRecords> newActivityConflictStatus = autoFulfillmentOrderRecordsMap.get(pageQueryForecastDetailResponse.getSalesOrderNo());
                            for (CfsBcAutoFulfillmentOrderRecords cb : newActivityConflictStatus) {
                                if (cb.getOrderLineBizNo().equals(pageQueryForecastDetailResponse.getLineNum())) {
                                    //同时信贷不足及库存不足时，分析原因只显示信贷不足，故信贷不足时需要再次判断
                                    if (new BigDecimal(pageQueryForecastDetailResponse.getPredictReplyAmount()).compareTo(new BigDecimal(pageQueryForecastDetailResponse.getQuantity())) < 0) {
                                        log.info("库存不足 , param:{}", JSON.toJSONString(pageQueryForecastDetailResponse));
                                        //重置circulation_status, 不能再次分析
                                        cb.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.STOCK_NOT_ENOUGH.getCode());
                                        cb.setFailReason(AutoFulfillmentOrderStatusCirculationEnum.STOCK_NOT_ENOUGH.getDesc());
                                        stockNotEnoughOrderBizNoList.add(cb.getOrderBizNo());
                                    } else {
                                        //信贷不足但库存充足时，重置circulation_status,不重新分析直接进行下一步：批量确认
                                        cb.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.ALL_PASS.getCode());
                                        cb.setFailReason("");
                                        needConfirmList.add(cb);
                                    }
                                }
                            }
                        }
                    }
                }
            }
            //只要有一个子单分析不满足，整个主单status都置为失败，且不需要确认
            if (CollectionUtils.isNotEmpty(stockNotEnoughOrderBizNoList)) {
                Map<String, List<CfsBcAutoFulfillmentOrderRecords>> groupMap = cfsBcAutoFulfillmentOrderRecordsList.stream().collect(Collectors.groupingBy(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo));
                for (String bizNo : stockNotEnoughOrderBizNoList) {
                    if (groupMap.containsKey(bizNo)) {
                        for (CfsBcAutoFulfillmentOrderRecords r : groupMap.get(bizNo)) {
                            r.setGmt_modified(new Date());
                            r.setStatus(AutoFulfillmentOrderStatusEnum.FAIL.getCode());
                            //防止needAnalyzeList有也包含该条数据，导致CirculationStatus被覆盖
                            Optional<CfsBcAutoFulfillmentOrderRecords> any = needAnalyzeList.stream().filter(cbfo -> cbfo.getId().equals(r.getId())).findAny();
                            if (any.isPresent()) {
                                r.setCirculationStatus(any.get().getCirculationStatus());
                            }
                            needUpdateList.add(r);
                        }
                    }
                    //需从待分析列表中删除
                    needAnalyzeList.removeIf(cbf -> cbf.getOrderBizNo().equals(bizNo));
                    //需要先更新到数据库，防止sendMessage时根据新的批次号查不到数据
                    for (CfsBcAutoFulfillmentOrderRecords needUpdate : needUpdateList) {
                        needUpdate.setGmt_modified(new Date());
                        autoFulfillmentOrderBizWriteService.update(needUpdate);
                    }
                }
                //调用发kafka消息方法，防止needAnalyzeList、needConfirmList、needSubmitList都为空，但是没有发kafka消息
                log.info("库存不足，发送kafka消息，has stockNotEnoughOrder,send kafka message");
                sendMessage(cfsBcAutoFulfillmentOrderRecordsList.get(0).getBatchNo(), cfsBcAutoFulfillmentOrderRecordsList.get(0).getFromSys(), cfsBcAutoFulfillmentOrderRecordsList.get(0).getBusinessType(), cfsBcAutoFulfillmentOrderRecordsList.get(0).getBusinessUnitCode());
            }

            //先更新status、批次号及请求号，发送kafka消息时校验状态需要用到(防止其它几种动作的status及批次号未更新)
            needUpdateList.addAll(needAnalyzeList);
            needUpdateList.addAll(needConfirmList);
            needUpdateList.addAll(needSubmitList);
            //此处更新操作必须要放在batchAnalyze之前，因为前面信贷不足重新判断了库存是否充足，重试时如果有子单分析完全满足后，会去查同主单下其它子单的状态是不是全部是完全满足或信贷不足但库存充足（仅仅是信贷不足时，也不用分析直接确认）
            for (CfsBcAutoFulfillmentOrderRecords needUpdate : needUpdateList) {
                needUpdate.setGmt_modified(new Date());
                autoFulfillmentOrderBizWriteService.update(needUpdate);
            }
            if (CollectionUtils.isNotEmpty(needAnalyzeList)) {
                log.info("重试，批量分析");
                //分析时，只能整单一起分析，因为可能单个子单分析的话都是全部满足，但整体一起分析的话又失败，比如单个分析库存是充足的，一起分析库存就不足；整体分析需较不分析比更加复杂
                QueryForListRequest reQueryRequest = new QueryForListRequest();
                reQueryRequest.setOrderBizNos(needAnalyzeList.stream().map(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo).distinct().collect(Collectors.toList()));
                reQueryRequest.setFromSys(needAnalyzeList.get(0).getFromSys());
                reQueryRequest.setBusinessType(needAnalyzeList.get(0).getBusinessType());
                reQueryRequest.setBusinessUnitCode(needAnalyzeList.get(0).getBusinessUnitCode());
                List<CfsBcAutoFulfillmentOrderRecords> reQueryResult = autoFulfillmentOrderBizQueryService.queryForList(reQueryRequest).stream().collect(Collectors.toList());
                batchAnalyze(reQueryResult);
            }
            Set<String> needAnalyzeOrderBizNoSet = needAnalyzeList.stream().map(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo).collect(Collectors.toSet());
            //如果同一主单有子单需要重新分析或已经判断是库存不足了，则该主单的其余子单不应该立即确认，需要等全部子单分析完成，所有子单满足进入确认阶段的条件，才能整体进行确认
            List<CfsBcAutoFulfillmentOrderRecords> afterFilterNeedConfirmList = needConfirmList.stream().filter(cfsRecord -> !needAnalyzeOrderBizNoSet.contains(cfsRecord.getOrderBizNo()) && !stockNotEnoughOrderBizNoList.contains(cfsRecord.getOrderBizNo())).collect(Collectors.toList());
            if (CollectionUtils.isNotEmpty(afterFilterNeedConfirmList)) {
                log.info("重试，批量确认");
                filterOrderAndBatchConfirm(afterFilterNeedConfirmList);
            }
            if (CollectionUtils.isNotEmpty(needSubmitList)) {
                log.info("重试，批量提交");
                batchSubmit(needSubmitList);
            }
        } catch (Exception e) {
            log.error("异步重试执行失败", e);
            throw e;
        }
        log.info("异步重试执行结束");
    }

    /**
     * 过滤出分析成功的订单，进行批量确认
     * @param cfsBcAutoFulfillmentOrderRecordsList 自动履约订单
     */
    @Override
    public void filterOrderAndBatchConfirm(List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsList) {
        log.info("批量确认开始, param:{}", JSON.toJSONString(cfsBcAutoFulfillmentOrderRecordsList));
        if (CollectionUtils.isEmpty(cfsBcAutoFulfillmentOrderRecordsList)) {
            return;
        }
        Map<String, List<CfsBcAutoFulfillmentOrderRecords>> orderBizNoMap = cfsBcAutoFulfillmentOrderRecordsList.stream().collect(Collectors.groupingBy(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo));
        ConfirmReplyRequest confirmReplyRequest = new ConfirmReplyRequest();
        String sequenceNo = customSequenceAccessor.getNextStringSequence("cr_forecast_sales", "sequenceNo");
        confirmReplyRequest.setTaskId(sequenceNo);
        confirmReplyRequest.setFromSys(cfsBcAutoFulfillmentOrderRecordsList.get(0).getFromSys());
        confirmReplyRequest.setBusinessType(cfsBcAutoFulfillmentOrderRecordsList.get(0).getBusinessType());
        confirmReplyRequest.setBusinessUnitCode(cfsBcAutoFulfillmentOrderRecordsList.get(0).getBusinessUnitCode());
        confirmReplyRequest.setModifierCode(cfsBcAutoFulfillmentOrderRecordsList.get(0).getUpdater());
        List<ConfirmForecastReplyDTO> confirmReplyList = new ArrayList<>();
        for (CfsBcAutoFulfillmentOrderRecords cfsBcAutoFulfillmentOrderRecords : cfsBcAutoFulfillmentOrderRecordsList) {
            ConfirmForecastReplyDTO confirmForecastReplyDTO = new ConfirmForecastReplyDTO();
            confirmForecastReplyDTO.setSalesOrderNo(cfsBcAutoFulfillmentOrderRecords.getOrderBizNo());
            confirmForecastReplyDTO.setLineNum(cfsBcAutoFulfillmentOrderRecords.getOrderLineBizNo());
            confirmReplyList.add(confirmForecastReplyDTO);
        }
        confirmReplyRequest.setConfirmReplyList(confirmReplyList);
        //批量确认预报订单
        log.info("批量确认带返回值开始 param:{}", JSON.toJSONString(confirmReplyRequest));
        List<ConfirmReplyResponseDTO> confirmReplyResponseDTOList = forecastOrderBizWriteService.batchConfirmWithReturn(confirmReplyRequest);
        log.info("批量确认带返回值结束 result:{}", JSON.toJSONString(confirmReplyResponseDTOList));
        List<CfsBcAutoFulfillmentOrderRecords> waitCreateSaleLineList = new ArrayList<>();
        for (ConfirmReplyResponseDTO confirmReplyResponseDTO : confirmReplyResponseDTOList) {
            String saleOrderNo = confirmReplyResponseDTO.getSalesOrderNo();
            if (orderBizNoMap.containsKey(saleOrderNo)) {
                List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordList = orderBizNoMap.get(saleOrderNo);
                for (CfsBcAutoFulfillmentOrderRecords autoFulfillmentOrderRecords : cfsBcAutoFulfillmentOrderRecordList) {
                    autoFulfillmentOrderRecords.setConfirmBatchNo(confirmReplyResponseDTO.getBatchTaskId());
                    autoFulfillmentOrderRecords.setGmt_modified(new Date());
                    if (confirmReplyResponseDTO.getSuccess()) {
                        autoFulfillmentOrderRecords.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.CREATE_SALE_ORDER_SUCCEED.getCode());
                        autoFulfillmentOrderRecords.setStatus(AutoFulfillmentOrderStatusEnum.PROCESSING.getCode());
                        autoFulfillmentOrderRecords.setFailReason("");
                        waitCreateSaleLineList.add(autoFulfillmentOrderRecords);
                    } else {
                        autoFulfillmentOrderRecords.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.CREATE_SALE_ORDER_FAILED.getCode());
                        autoFulfillmentOrderRecords.setStatus(AutoFulfillmentOrderStatusEnum.FAIL.getCode());
                        autoFulfillmentOrderRecords.setFailReason(confirmReplyResponseDTO.getMessage());
                    }
                    autoFulfillmentOrderBizWriteService.update(autoFulfillmentOrderRecords);
                }
            }
        }
        boolean allFailed = confirmReplyResponseDTOList.stream().allMatch(confirmReplyResponseDTO -> !confirmReplyResponseDTO.getSuccess());
        if (CollectionUtils.isNotEmpty(waitCreateSaleLineList)) {
            //发送rocketMQ消息，用于检查是否所有销售子单都已确认结束
            String message = JSON.toJSONString(waitCreateSaleLineList.get(0));
            Map<String, String> map = new HashMap<>(1);
            map.put("_CZ_KEY", BcOrderConstants.TENANT_ID());
            Boolean sendResult = messageService.sendMessage("auto_fulfillment_message", "CR_AUTO_FULFILLMENT_ORDER", "AUTO_FULFILLMENT_ORDER_PUSH", waitCreateSaleLineList.get(0).getConfirmBatchNo(), message, map);
            log.info("AutoFulfillmentOrderServiceImpl#filterOrderAndBatchConfirm AUTO_FULFILLMENT_ORDER_PUSH message:{}, sendResult:{}", message, sendResult);
        } else if (allFailed) {
            log.info("forecastOrderBizWriteService.batchConfirmWithReturn all failed, send kafka message");
            sendMessage(cfsBcAutoFulfillmentOrderRecordsList.get(0).getBatchNo(), cfsBcAutoFulfillmentOrderRecordsList.get(0).getFromSys(), cfsBcAutoFulfillmentOrderRecordsList.get(0).getBusinessType(), cfsBcAutoFulfillmentOrderRecordsList.get(0).getBusinessUnitCode());
        }
        log.info("批量确认执行结束");
    }

    /**
     * 批量提交（销售订单）
     * @param needSubmitList
     */
    private void batchSubmit(List<CfsBcAutoFulfillmentOrderRecords> needSubmitList) {
        log.info("批量提交开始, param:{}", JSON.toJSONString(needSubmitList));
        if (CollectionUtils.isEmpty(needSubmitList)) {
            return;
        }
        String sequenceNo = customSequenceAccessor.getNextStringSequence("cr_forecast_sales", "sequenceNo");
        OrderAtomBatchSubmitRequest orderAtomBatchSubmitRequest = new OrderAtomBatchSubmitRequest();
        orderAtomBatchSubmitRequest.setReqNo(sequenceNo);
        orderAtomBatchSubmitRequest.setFromSys(needSubmitList.get(0).getFromSys());
        orderAtomBatchSubmitRequest.setBusinessType(needSubmitList.get(0).getBusinessType());
        orderAtomBatchSubmitRequest.setBusinessUnitCode(needSubmitList.get(0).getBusinessUnitCode());
        List<OrderAtomSubmitRequest> orderList = new ArrayList<>();
        for (CfsBcAutoFulfillmentOrderRecords record : needSubmitList) {
            OrderAtomSubmitRequest orderAtomSubmitRequest = new OrderAtomSubmitRequest();
            orderAtomSubmitRequest.setReqNo(orderAtomBatchSubmitRequest.getReqNo());
            orderAtomSubmitRequest.setFromSys(orderAtomBatchSubmitRequest.getFromSys());
            orderAtomSubmitRequest.setBusinessType(orderAtomBatchSubmitRequest.getBusinessType());
            orderAtomSubmitRequest.setBusinessUnitCode(orderAtomBatchSubmitRequest.getBusinessUnitCode());
            orderAtomSubmitRequest.setCheckCredit(Boolean.FALSE);
            orderAtomSubmitRequest.setCheckStock(Boolean.FALSE);
            orderAtomSubmitRequest.setModifierCode(record.getUpdater());
            Modifiers modifier = invalidModifier(record.getUpdater());
            orderAtomSubmitRequest.setModifierName(modifier.getName());
            if (StringUtils.isNotEmpty(record.getFeatures())) {
                JSONObject jsonObject = JSONObject.parseObject(record.getFeatures());
                if (jsonObject.containsKey("salesOrderNo")) {
                    orderAtomSubmitRequest.setSalesOrderNo((String) jsonObject.get("salesOrderNo"));
                }
                if (jsonObject.containsKey("outOrderNo")) {
                    orderAtomSubmitRequest.setOutOrderNo((String) jsonObject.get("outOrderNo"));
                }
            }
            orderList.add(orderAtomSubmitRequest);
        }
        //orderList根据salesOrderNo去下重
        List<OrderAtomSubmitRequest> distinctOrderAtomSubmitRequest = orderList.stream().collect(
                Collectors.collectingAndThen(Collectors.toCollection(() -> new TreeSet<>(Comparator.comparing(OrderAtomSubmitRequest::getSalesOrderNo))),
                        ArrayList::new));
        orderAtomBatchSubmitRequest.setOrderList(distinctOrderAtomSubmitRequest);
        log.info("重试，批量提交销售单开始 param:{}", JSON.toJSONString(orderAtomBatchSubmitRequest));
        Result<List<OrderAtomSubmitResponse>> batchSubmitResult = salesOrderAtomService.batchSubmit(orderAtomBatchSubmitRequest);
        log.info("重试，批量提交销售单结束 result:{}", JSON.toJSONString(batchSubmitResult));
        if (batchSubmitResult.isSuccess()) {
            for (CfsBcAutoFulfillmentOrderRecords needSubmit : needSubmitList) {
                needSubmit.setStatus(AutoFulfillmentOrderStatusEnum.PROCESSING.getCode());
                needSubmit.setGmt_modified(new Date());
                autoFulfillmentOrderBizWriteService.update(needSubmit);
            }
            //发送rocketMQ消息，用于检查是否所有销售子单都已推送sap结束
            String message = JSON.toJSONString(needSubmitList.get(0));
            Map<String, String> map = new HashMap<>(1);
            map.put("_CZ_KEY", BcOrderConstants.TENANT_ID());
            Boolean sendResult = messageService.sendMessage("auto_fulfillment_message", "CR_AUTO_FULFILLMENT_ORDER", "AUTO_FULFILLMENT_ORDER_PUSH", needSubmitList.get(0).getConfirmBatchNo(), message, map);
            log.info("AutoFulfillmentOrderServiceImpl#asyncRetry retry AUTO_FULFILLMENT_ORDER_PUSH message:{}, sendResult:{}", message, sendResult);
        } else {
            log.info("重试，批量提交执行异常");
            throw new FunctionException("OC-04-001-01-16-033");
        }
    }

    @Override
    public Result<Void> sendMessage(String batchNo, String fromSys, String businessType, String businessUnitCode) {
        log.info("发送Kafka消息开始, batchNo:{}, fromSys:{}, businessType:{}, businessUnitCode:{}", batchNo, fromSys, businessType, businessUnitCode);
        try {
            QueryForListRequest queryForListRequest = new QueryForListRequest();
            queryForListRequest.setBatchNo(batchNo);
            queryForListRequest.setFromSys(fromSys);
            queryForListRequest.setBusinessType(businessType);
            queryForListRequest.setBusinessUnitCode(businessUnitCode);
            List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecords = autoFulfillmentOrderBizQueryService.queryForList(queryForListRequest);
            if (CollectionUtils.isEmpty(cfsBcAutoFulfillmentOrderRecords)) {
                log.info("kafka sendMessage no data find ,param:{} ", JSON.toJSONString(queryForListRequest));
                return Result.success(null);
            }
            boolean hasProcessing = !cfsBcAutoFulfillmentOrderRecords.stream().filter(cfsOrderRecord -> AutoFulfillmentOrderStatusEnum.PROCESSING.getCode() == cfsOrderRecord.getStatus()).collect(Collectors.toList()).isEmpty();
            if (hasProcessing) {
                log.info("有状态为processing，流程未结束，跳过，不发送消息,param:{} ", JSON.toJSONString(queryForListRequest));
                return Result.success(null);
            }
            List<AutoFulfillmentOrderDTO> autoFulfillmentOrders = new ArrayList<>();
            Map<String, List<CfsBcAutoFulfillmentOrderRecords>> groupBySourceOrderNo = cfsBcAutoFulfillmentOrderRecords.stream().collect(Collectors.groupingBy(CfsBcAutoFulfillmentOrderRecords::getSourceOrderNo));
            for (Map.Entry<String, List<CfsBcAutoFulfillmentOrderRecords>> entry : groupBySourceOrderNo.entrySet()) {
                AutoFulfillmentOrderDTO autoFulfillmentOrderDTO = new AutoFulfillmentOrderDTO();
                List<AutoFulfillmentOrderDetailDTO> autoFulfillmentOrderDetails = new ArrayList<>();
                Integer status = 1;
                StringBuilder messageBuilder = new StringBuilder();
                for (CfsBcAutoFulfillmentOrderRecords cfsOrder : entry.getValue()) {
                    AutoFulfillmentOrderDetailDTO autoFulfillmentOrderDetailDTO = new AutoFulfillmentOrderDetailDTO();
                    autoFulfillmentOrderDetailDTO.setOrderLineBizNo(cfsOrder.getOrderLineBizNo());
                    autoFulfillmentOrderDetailDTO.setSourceOrderLineNo(cfsOrder.getSourceOrderLineNo());
                    autoFulfillmentOrderDetailDTO.setSapOrderLineNo(cfsOrder.getSapOrderLineNo());
                    if (AutoFulfillmentOrderStatusEnum.FAIL.getCode() == cfsOrder.getStatus()) {
                        status = 0;
                        if (AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getCode() == cfsOrder.getCirculationStatus()
                                || AutoFulfillmentOrderStatusCirculationEnum.ANALYZE_FAILED.getCode() == cfsOrder.getCirculationStatus()
                                || AutoFulfillmentOrderStatusCirculationEnum.CREATE_SALE_ORDER_FAILED.getCode() == cfsOrder.getCirculationStatus()
                                || AutoFulfillmentOrderStatusCirculationEnum.PUSH_TO_SAP_FAILED.getCode() == cfsOrder.getCirculationStatus()) {
                            messageBuilder.setLength(0);
                            messageBuilder.append(cfsOrder.getFailReason()).append("; ");
                        } else {
                            messageBuilder.append(cfsOrder.getSourceOrderLineNo()).append(cfsOrder.getFailReason()).append("; ");
                        }
                    }
                    autoFulfillmentOrderDetails.add(autoFulfillmentOrderDetailDTO);
                    autoFulfillmentOrderDTO.setOrderBizNo(cfsOrder.getOrderBizNo());
                    autoFulfillmentOrderDTO.setSourceOrderNo(cfsOrder.getSourceOrderNo());
                    autoFulfillmentOrderDTO.setSapOrderNo(cfsOrder.getSapOrderNo());
                }
                autoFulfillmentOrderDTO.setAutoFulfillmentOrderDetails(autoFulfillmentOrderDetails);
                autoFulfillmentOrderDTO.setStatus(String.valueOf(status));
                autoFulfillmentOrderDTO.setMessage(messageBuilder.toString());
                autoFulfillmentOrders.add(autoFulfillmentOrderDTO);
            }
            CfsBcAutoFulfillmentOrderRecords orderRecord = cfsBcAutoFulfillmentOrderRecords.get(0);
            CommonAsyncMessageResponse response = buildCommonAsyncMessageResponse(autoFulfillmentOrders, orderRecord.getFromSys(), orderRecord.getBusinessUnitCode(), AsyncMessageEventEnum.ORDER_COMPLETED.getValue(),
                    AsyncMessageEventEnum.ORDER_COMPLETED.getLabel(), orderRecord.getBatchNo(), orderRecord.getReqNo());
            //返回异步消息
            log.info("send auto fulfillment order message to mn request:" + JSON.toJSONString(response));
            String messageKey = KafkaUtils.generateMessageKey(fromSys, "", businessUnitCode, AsyncMessageEventEnum.ORDER_COMPLETED.getValue());
            SendResult<Integer, String> sendResult = KafkaUtils.send(MnOrderCenterConstants.AUTO_FULFILLMENT_ORDER_KAFKA_TOPIC(), messageKey, JSON.toJSONString(response));
            log.info("send auto fulfillment order message to mn result: {}", sendResult);
            return Result.success(null);
        } catch (Exception e) {
            log.error("send kafka message failed.", e);
            throw e;
        }
    }

    private CommonAsyncMessageResponse buildCommonAsyncMessageResponse(List<AutoFulfillmentOrderDTO> autoFulfillmentOrders, String fromSys, String businessUnitCode,
                                                                       String eventCode, String eventName, String batchNo, String reqNo) {
        CommonAsyncMessageResponse commonAsyncMessageResponse = new CommonAsyncMessageResponse();
        commonAsyncMessageResponse.setEventCode(eventCode);
        commonAsyncMessageResponse.setEventName(eventName);
        commonAsyncMessageResponse.setEventTime(String.valueOf(System.currentTimeMillis()));
        commonAsyncMessageResponse.setFromSys(fromSys);
        commonAsyncMessageResponse.setBusinessUnitCode(businessUnitCode);
        commonAsyncMessageResponse.setBatchNo(batchNo);
        commonAsyncMessageResponse.setReqNo(reqNo);
        commonAsyncMessageResponse.setData(autoFulfillmentOrders);
        return commonAsyncMessageResponse;
    }

    private List<PageQueryForecastDetailResponse> pageQueryLine(List<String> orderBizNoList, CfsBcAutoFulfillmentOrderRecords cfs) {
        List<PageQueryForecastDetailResponse> result = new ArrayList<>();
        boolean hasNext = true;
        int pageNo = 1;
        PageQueryForecastLineRequest pageQueryForecastLineRequest = new PageQueryForecastLineRequest();
        pageQueryForecastLineRequest.setSalesOrderNoList(orderBizNoList);
        pageQueryForecastLineRequest.setSize(100);
        pageQueryForecastLineRequest.setFromSys(cfs.getFromSys());
        pageQueryForecastLineRequest.setBusinessType(cfs.getBusinessType());
        pageQueryForecastLineRequest.setBusinessUnitCode(cfs.getBusinessUnitCode());
        while (hasNext) {
            pageQueryForecastLineRequest.setPage(pageNo);
            log.info("查询预报子单行开始 param:{}", JSON.toJSONString(pageQueryForecastLineRequest));
            Result<List<PageQueryForecastDetailResponse>> forecastOrderDetailQueryResult = forecastOrderBizDomainService.pageQueryLine(pageQueryForecastLineRequest);
            log.info("查询预报子单行结束 success:{}", forecastOrderDetailQueryResult.isSuccess());
            if (forecastOrderDetailQueryResult.isSuccess() && CollectionUtils.isNotEmpty(forecastOrderDetailQueryResult.getResult())) {
                result.addAll(forecastOrderDetailQueryResult.getResult());
            }
            if (forecastOrderDetailQueryResult.getTotal() <= result.size()) {
                hasNext = false;
            }
            pageNo++;
            if (pageNo >= 20) {
                break;
            }
        }
        return result;
    }


}
